package com.niit.spark.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
 * Date:2025/5/21
 * Author：Ys
 * Description:
 */
object SparkStreaming03_Kafka {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming02_Queue")
    val ssc = new StreamingContext(conf,Seconds(5))
    ssc.sparkContext.setLogLevel("ERROR")

    //创建Map,里面存储Kafka的配置信息
    val kafkaMap = Map[String,Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "node1:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "SP_KF",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer].getName,
    )

    //读取Kafka数据创建DStream
    val kafkaDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Set("BD1"), kafkaMap))

    val infoDS: DStream[String] = kafkaDS.map(record => {

      val topic: String = record.topic()
      val value: String = record.value()
      val key: String = record.key()
      val partition: Int = record.partition()
      val offset: Long = record.offset()
      val info = s"topic:$topic,value:$value,key:$key,partition:$partition,offset:$offset"
      info

    })

    infoDS.print()


    ssc.start()
    ssc.awaitTermination()
  }

}
